package com.yyfax.pay.quartz.jobs;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.commons.collections.CollectionUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.yyfax.framework.common.Log;
import com.yyfax.framework.spring.ApplicationContextHolder;
import com.yyfax.pay.common.constants.QuartzConstant;
import com.yyfax.pay.common.enums.TaskStateEnum;
import com.yyfax.pay.depository.dao.FuiouQueryTaskDao;
import com.yyfax.pay.depository.domain.FuiouQueryTask;
import com.yyfax.pay.quartz.service.FuiouDepositQueryService;

/**
 * 富友存管-主动查询分派（扫描机制）
 * 
 * @author buyi	
 * @date 2017-09-12 11:36:30
 * @since v2.0.8
 */
@Component
public class FuiouDepositQueryFixJobDispatcher implements Job {
	private static final Logger logger = LoggerFactory.getLogger(FuiouDepositQueryFixJobDispatcher.class);

	private static FuiouQueryTaskDao fuiouQueryTaskDao;

	private static FuiouDepositQueryService fuiouDepositQueryService;

	@Resource
	private QuartzConstant quartzConstant;

	/**
	 * 连接池
	 */
	private ExecutorService executors;

	@PostConstruct
	private void init() {

		fuiouQueryTaskDao = ApplicationContextHolder.getBean(FuiouQueryTaskDao.class);
		fuiouDepositQueryService = ApplicationContextHolder.getBean(FuiouDepositQueryService.class);

		// 主要为IO请求，线程池大小设置为2*CPU个数+1
		int coreSize = Runtime.getRuntime().availableProcessors() * 2 + 1;
		this.executors = new ThreadPoolExecutor(coreSize, quartzConstant.getQuartzPoolMaxSize(), quartzConstant.getQuartzPoolKeepAlive(),
				TimeUnit.HOURS, new LinkedBlockingQueue<Runnable>(quartzConstant.getQuartzPoolQueueSize()),
				new FuioudepositQueryRejectedExecutionHandler());
	}

	@Override
	public void execute(JobExecutionContext context) throws JobExecutionException {
		String op = "FuiouDepositQueryFixJobDispatcher.execute";
		logger.info(Log.op(op).msg("执行查询任务").kv("context", context).toString());

		List<FuiouQueryTask> tasks = fuiouQueryTaskDao.selectActiveList(quartzConstant.getQueryPageSize());
		if (CollectionUtils.isEmpty(tasks)) {
			logger.info(Log.op(op).msg("【富友存管】【定时查询】没有可执行记录").toString());
			return;
		}

		// 抽取主鍵集合
		List<Integer> ids = Lists.transform(tasks, new Function<FuiouQueryTask, Integer>() {
			@Override
			public Integer apply(FuiouQueryTask input) {
				return input.getId();
			}
		});

		// 更新任务状态为处理中
		fuiouQueryTaskDao.updateStateList(TaskStateEnum.DEALING.getValue(), ids);

		handleList(tasks);
	}

	/**
	 * 执行批量任务集合
	 * 
	 * @author buyi
	 * @date 2018-01-23 10:41:32
	 * @since v2.0.8
	 * @param tasks
	 */
	private void handleList(List<FuiouQueryTask> tasks) {
		for (FuiouQueryTask task : tasks) {
			executors.execute(FuioudepositQueryThread.newInstance(task));
		}
	}

	/**
	 * 查询线程
	 * 
	 * @author buyi
	 * @date 2018-01-29 09:36:05
	 * @since v2.0.8
	 */
	private static class FuioudepositQueryThread implements Runnable {
		private FuiouQueryTask task;

		public FuioudepositQueryThread(FuiouQueryTask task) {
			super();
			this.task = task;
		}

		public static FuioudepositQueryThread newInstance(FuiouQueryTask task) {
			return new FuioudepositQueryThread(task);
		}

		public FuiouQueryTask getTask() {
			return task;
		}

		@Override
		public void run() {
			fuiouDepositQueryService.executeTask(task);
		}
	}

	/**
	 * 查询连接加入连接池超时拒绝处理
	 * 
	 * @author buyi
	 * @date 2018-01-29 09:35:25
	 * @since v2.0.8
	 */
	private static class FuioudepositQueryRejectedExecutionHandler implements RejectedExecutionHandler {
		private static final Logger logger = LoggerFactory.getLogger(FuioudepositQueryRejectedExecutionHandler.class);

		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			final String op = "FuioudepositQueryRejectedExecutionHandler.rejectedExecution";
			if (!(r instanceof FuioudepositQueryThread)) {
				// 如果线程类型不是富友存管通知类型，则不执行
				logger.warn(Log.op(op).msg("【富友存管】【定时查询】添加异步通知任务到线程池，任务队列已满，线程类型非指定FuioudepositQueryThread类型").toString());
				return;
			}

			FuioudepositQueryThread thread = (FuioudepositQueryThread) r;
			FuiouQueryTask task = thread.getTask();

			logger.warn(Log.op(op).msg("【富友存管】【定时查询】添加异步通知任务到线程池，任务队列已满，重置任务状态").kv("id", task.getId()).kv("busiId", task.getBusiId()).toString());

			// 如果添加线程池失败，则更新任务状态为初始化状态，等待下一次轮询
			fuiouQueryTaskDao.updateStateById(task.getId(), TaskStateEnum.ACTIVE.getValue());

		}
	}
}
